package adapter

import (
	"context"
	"fmt"
	"gitee.com/zr233/gqueue/errors"
	"github.com/Shopify/sarama"
	"time"
)

//key: [topic-groupID]
type consumerMap map[string]*ConsumerOrigin

//key: [topic] - value: []*ConsumerOrigin
type topicConsumerMap map[string][]*ConsumerOrigin

func (c *consumerMap) genKey(topic, groupId string) string {
	return fmt.Sprintf("%s-%s", topic, groupId)
}
func (t *topicConsumerMap) genKey(topic string) string {
	return topic
}

type Origin struct {
	cfg *OriginConfig

	consumers          consumerMap
	topicConsumerIndex topicConsumerMap
	base
}

func (o *Origin) Close() error {
	o.cancel()
	return nil
}

func (o *Origin) NewProducer() (p Producer, err error) {
	p = NewProducerOrigin(o)
	return
}

type OriginConfig struct {
	QueueLen int
}

func NewOriginConfig() *OriginConfig {
	o := &OriginConfig{
		QueueLen: 50000,
	}
	return o
}

func NewOrigin(cfg *OriginConfig) (o *Origin, err error) {
	o = &Origin{
		consumers:          consumerMap{},
		topicConsumerIndex: topicConsumerMap{},
		cfg:                cfg,
	}
	o.init()

	return
}

type ProducerOrigin struct {
	core      *Origin
	inputChan chan *sarama.ProducerMessage
	errChan   chan *sarama.ProducerError
}

func (p *ProducerOrigin) Errors() <-chan *sarama.ProducerError {
	return p.errChan
}

func NewProducerOrigin(core *Origin) *ProducerOrigin {
	p := &ProducerOrigin{
		core:      core,
		inputChan: make(chan *sarama.ProducerMessage, 100),
		errChan:   make(chan *sarama.ProducerError, 100),
	}
	go func() {
		for {
			if core.ctx.Err() != nil {
				return
			}
			p.dealInput()
		}
	}()

	return p
}
func (p *ProducerOrigin) reportError(m *sarama.ProducerMessage, err error) {
	pErr := &sarama.ProducerError{
		Msg: m,
		Err: err,
	}
	p.errChan <- pErr
}
func (p *ProducerOrigin) dealInput() {
	select {
	case <-p.core.ctx.Done():
		return
	case m := <-p.inputChan:
		consumerList := p.core.topicConsumerIndex[m.Topic]

		mNew := &sarama.ConsumerMessage{
			Timestamp:      m.Timestamp,
			BlockTimestamp: time.Time{},
			Topic:          m.Topic,
		}
		var err error

		if m.Key != nil {
			mNew.Key, err = m.Key.Encode()
			if err != nil {
				p.reportError(m, err)
				return
			}
		}
		if m.Value != nil {
			mNew.Value, err = m.Value.Encode()
			if err != nil {
				p.reportError(m, err)
				return
			}
		}

		for _, consumer := range consumerList {
			select {
			case consumer.outputChan <- mNew:
			case <-p.core.ctx.Done():
				return
			default:
				err = fmt.Errorf("[%s]%w", consumer.groupId, errors.ErrQueueFull)
				p.reportError(m, err)
				continue
			}
		}
	}
}

func (p *ProducerOrigin) Input() chan<- *sarama.ProducerMessage {
	return p.inputChan
}

type ConsumerOrigin struct {
	topic      string
	groupId    string
	core       *Origin
	outputChan chan *sarama.ConsumerMessage
}

func (o *Origin) NewConsumer(topic, groupId string) (c Consumer, err error) {
	newConsumer := &ConsumerOrigin{
		core:       o,
		topic:      topic,
		groupId:    groupId,
		outputChan: make(chan *sarama.ConsumerMessage, o.cfg.QueueLen),
	}
	consumerKey := o.consumers.genKey(topic, groupId)
	o.consumers[consumerKey] = newConsumer
	indexKey := o.topicConsumerIndex.genKey(topic)
	o.topicConsumerIndex[indexKey] = append(o.topicConsumerIndex[indexKey], newConsumer)

	c = newConsumer
	return
}

func (c *ConsumerOrigin) Get() <-chan *sarama.ConsumerMessage {
	return c.outputChan
}

func (c *ConsumerOrigin) GetBatch(ctx context.Context, batch int) (list []*sarama.ConsumerMessage) {
	for i := 0; i < batch; i++ {
		select {
		case <-c.core.ctx.Done():
			return
		case <-ctx.Done():
			return
		case m := <-c.outputChan:
			list = append(list, m)
		}
	}
	return
}
